package com.sky.kafka.producer.web;

import com.alibaba.fastjson.JSONObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("producer")
public class ProducerController {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @GetMapping("producer")
    public String producer(String name) {
        JSONObject json = new JSONObject();
        json.put("name", name);
        json.put("company", "HCT");
        kafkaTemplate.send("name", json.toJSONString());
        System.out.println(name);
        return name;
    }

    @GetMapping("name")
    public String name(String addr) {
        JSONObject json = new JSONObject();
        json.put("addr", addr);
        json.put("name", "Sky");
        kafkaTemplate.send("sky", json.toJSONString());
        System.out.println(addr);
        return addr;
    }

    @GetMapping("poll")
    public String poll(){
        for (int i = 1; i < 11; i++) {
            kafkaTemplate.send("poll", "测试拉取消费" + i);
        }
        return "发送完成";
    }
}
